package com.cwrl.spider.common.mq.config;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.util.TypeUtil;
import com.alibaba.fastjson.JSON;
import com.cwrl.spider.common.mq.annotaion.SpiderMqListener;
import com.cwrl.spider.common.mq.core.ISpiderMqListener;
import com.cwrl.spider.common.mq.domain.SpiderMqMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.util.*;

/**
 * @author cw
 */
@Slf4j
public class SpiderMqBeanPostProcessor implements ApplicationContextAware,BeanPostProcessor, SmartInitializingSingleton {

    private final Set<Object> objects = new HashSet<>();
    private ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        for (Method method : bean.getClass().getMethods()) {
            boolean hasAnnotation = method.isAnnotationPresent(SpiderMqListener.class);
            if (hasAnnotation) {
                objects.add(bean);
            }
        }
        return bean;
    }

    @Override
    public void afterSingletonsInstantiated() {

        RedisMessageListenerContainer container = applicationContext.getBean(RedisMessageListenerContainer.class);
        applicationContext.getBeansOfType(ISpiderMqListener.class).forEach((key, listener) -> {
            SpiderMqListener annotation = listener.getClass().getAnnotation(SpiderMqListener.class);
            if (Objects.isNull(annotation)) {
                return;
            }

            List<ChannelTopic> topicList = new ArrayList<>();
            for (String topic: annotation.topic()) {
                topicList.add(new ChannelTopic(topic));
            }

            container.addMessageListener((message, mess) -> {
                try {
                    Type type = TypeUtil.getTypeArgument(listener.getClass(), 0);
                    SpiderMqMessage<?> spiderMqMessage = JSON.parseObject(message.getBody(), SpiderMqMessage.class);
                    listener.onMessage(Convert.convert(type,spiderMqMessage.getMessage()));
                } catch (Exception e) {
                    log.error("处理消息异常", e);
                }

            }, topicList);
        });

        for (Object bean : objects) {
            for (Method method : bean.getClass().getMethods()) {
                SpiderMqListener methodAnnotation = method.getAnnotation(SpiderMqListener.class);
                if (methodAnnotation == null) {
                    continue;
                }
                List<ChannelTopic> methodTopicList = new ArrayList<>();
                for (String topic : methodAnnotation.topic()) {
                    methodTopicList.add(new ChannelTopic(topic));
                }
                // 包含此注解
                container.addMessageListener((message, mess) -> {
                    Object[] params = new Object[method.getParameterCount()];
                    SpiderMqMessage<?> spiderMqMessage = JSON.parseObject(message.getBody(), SpiderMqMessage.class);
                    try {
                        if (params.length > 0) {
                            Class<?> parameterType = method.getParameterTypes()[0];
                            params[0] = Convert.convert(parameterType, spiderMqMessage.getMessage());
                        }
                        method.invoke(bean, params);
                    } catch (Exception e) {
                        log.error("处理消息异常", e);
                    }
                }, methodTopicList);
            }
        }
    }


}
